package sql;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.sql.Timestamp;

/**

 双流join

 */
public class D10_1datagen_window_join {


    public static void main(String[] args) {

        System.out.println(Integer.MAX_VALUE);
        System.out.println(Long.MAX_VALUE);
        Configuration flinkConf = new Configuration();
        flinkConf.setString("rest.port","9091");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(flinkConf);
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        String genSqlCpu = "CREATE TABLE ods_tb_cpu ( " +
            " stime TIMESTAMP(3)," +
            " sn STRING," +
            " cpu BIGINT," +
            " rowtime AS localtimestamp," +
            " WATERMARK FOR rowtime AS rowtime - interval '10' second" +
            ") WITH ( " +
            " 'connector'='datagen', " +
            " 'rows-per-second'='1'," +
            " 'fields.sn.length'='1'," +
            " 'fields.cpu.kind'='random'," +
            " 'fields.cpu.min'='0'," +
            " 'fields.cpu.max'='100'" +
            ")";

        String genSqlMem = "CREATE TABLE ods_tb_mem ( " +
            " stime TIMESTAMP(3)," +
            " sn STRING," +
            " mem BIGINT," +
            " rowtime AS localtimestamp," +
            " WATERMARK FOR rowtime AS rowtime - interval '10' second" +
            ") WITH ( " +
            " 'connector'='datagen', " +
            " 'rows-per-second'='1'," +
            " 'fields.sn.length'='1'," +
            " 'fields.mem.kind'='random'," +
            " 'fields.mem.min'='0'," +
            " 'fields.mem.max'='100'" +
            ")";


        String print = "CREATE TABLE print (" +"    " +
            "    window_start timestamp, " +
            "    window_end timestamp, " +
            "    sn STRING, " +
            "    cpu BIGINT, " +
            "    mem BIGINT " +
            ") WITH (" +
            "     'connector' = 'print'" +
            ")";

        String sql = "insert into print " +
                " SELECT " +
            "    cpu.window_start AS window_start, " +
            "    cpu.window_end AS window_end, " +
            "    ifNull(ifNull(cpu.sn, mem.sn), cpu.sn) as sn," +
            "    cpu, " +
            "    mem " +
            "FROM\n" +
            "    (\n" +
            "        SELECT * FROM TABLE(tumble(TABLE ods_tb_cpu, descriptor(rowtime), INTERVAL '1' minutes))\n" +
            "    ) AS cpu\n" +
            "    FULL JOIN (\n" +
            "        SELECT * FROM TABLE(tumble(TABLE ods_tb_mem, descriptor(rowtime), INTERVAL '1' minutes))\n" +
            "    ) AS mem \n" +
            "    ON cpu.sn = mem.sn\n" +
            "    AND cpu.window_start = mem.window_start\n" +
            "    AND cpu.window_end = mem.window_end";

        /**
         *
         */




        tableEnv.executeSql(genSqlCpu);
        tableEnv.executeSql(genSqlMem);

        tableEnv.executeSql("desc ods_tb_cpu").print();
        tableEnv.executeSql("desc ods_tb_mem").print();

        tableEnv.executeSql(print);

        System.out.println(sql);
        tableEnv.executeSql(sql);


        System.out.println(new Timestamp(System.currentTimeMillis()));



    }
}
